package net.lb.socketserver.handler;

import java.util.Arrays;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import net.lb.baseutil.order.AescComparator;
import net.lb.baseutil.order.DescComparator;
import net.lb.socketserver.bus.Message;
import net.lb.socketserver.interceptor.MessageInterceptor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * 消息动作类，完成对消息的进入拦截，请求处理，出口拦截，写出到通道。实现Callable接口，放入线程池进行调用。
 * 
 * @author linbin
 * 
 */
public final class MessageAction implements Callable<Void>
{
    private Queue<Message>       queue  = new ConcurrentLinkedQueue<>();
    private Logger               logger = LogManager.getLogger();
    private MessageInterceptor[] inInterceptors;
    private MessageInterceptor[] outInterceptors;
    private int                  interceptorNum;
    
    /**
     * 使用拦截器数组初始化对象，会自动将拦截器进行排序，按照入口顺序和出口顺序存放
     * 
     * @param interceptors
     */
    public MessageAction(MessageInterceptor[] interceptors)
    {
        this.inInterceptors = interceptors;
        Arrays.sort(this.inInterceptors, new AescComparator());
        interceptorNum = inInterceptors.length;
        outInterceptors = Arrays.copyOf(inInterceptors, interceptorNum);
        Arrays.sort(outInterceptors, new DescComparator());
    }
    
    @Override
    public Void call() throws Exception
    {
        Message msg = queue.poll();
        try
        {
            int index = 0;
            int result = -1;
            for (; index < interceptorNum;)
            {
                result = inInterceptors[index].inInterceptor(msg);
                index++;
                if (result > 1)
                {
                    break;
                }
            }
            if (result == MessageInterceptor.closed)
            {
                msg.getChannelInfo().close();
                return null;
            }
            if (result != MessageInterceptor.interceptored)
            {
                msg.getHandler().handler(msg);
            }
            for (int i = interceptorNum - index; i < interceptorNum; i++)
            {
                result = outInterceptors[i].outInterceptor(msg);
                if (result > 1)
                {
                    break;
                }
            }
            if (result == MessageInterceptor.closed)
            {
                msg.getChannelInfo().close();
                return null;
            }
            msg.flowFinish();
            if (msg.getChannelInfo().isOpen())
            {
                msg.tryWrite();
            }
            else
            {
                logger.debug("通道{}已经关闭，不发送消息", msg.getChannelInfo().getAddress());
            }
        }
        catch (Exception e)
        {
            logger.error("消息处理过程发生异常，关闭通道，当前buffer状态", msg.getWriteBuffer(), e);
            msg.getChannelInfo().close();
        }
        return null;
    }
    
    public void addMessage(Message message)
    {
        queue.add(message);
    }
    
}
